#ifndef __M_CONNEXTION_H__
#define __M_CONNEXTION_H__
#include "muduo/proto/dispatcher.h"
#include "muduo/proto/codec.h"


#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpClient.h"
#include"muduo/net/EventLoopThread.h"
#include"muduo/base/CountDownLatch.h"
#include<iostream>

#include"mq_channel.hpp"
#include"mq_work.hpp"
namespace bitmq{
class Connection{
    public:
    typedef std::shared_ptr<google::protobuf::Message> MessagePtr;
        using ptr=std::shared_ptr<Connection>;
        Connection(const std::string &sip, int sport,const AsyncWord::ptr& work): _latch(1),
         _connection(work->_loopthread.startLoop(), muduo::net::InetAddress(sip, sport), "Connection"),
            _dispatcher(std::bind(&Connection::onUnknownMessage, this, std::placeholders::_1, 
                std::placeholders::_2, std::placeholders::_3)),
            _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),
                _channel_manager(std::make_shared<bitmq::ChannelManager>()),
                _work(work)
                {

            _dispatcher.registerMessageCallback<basicCommonRespose>(std::bind(&Connection::basicResponse, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
                
            _dispatcher.registerMessageCallback<basicConsumerRespose>(std::bind(&Connection::consumerResponse, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            _connection.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(),
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _connection.setConnectionCallback(std::bind(&Connection::onConnection, this, std::placeholders::_1));      

              _connection.connect();
            _latch.wait();//阻塞等待，直到连接建立成功
        }
   
   
    public:
    Channel::ptr createChannel()
    {
        Channel::ptr channel=_channel_manager->create(_conn,_codec);
        bool ret=channel->openChannel();
        if(ret==false)
        {
            DLOG("服务端打开信道失败!");
        }
        
        return channel;
    }
    void closeChannel(const Channel::ptr&channel)
    {
        channel->closeChannel(channel->cid());
        _channel_manager->remove(channel->cid());
    }
    public:
    void onUnknownMessage(const muduo::net::TcpConnectionPtr& conn, const MessagePtr& message, muduo::Timestamp) {
       LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
        conn->shutdown();
    }
    void basicResponse(const muduo::net::TcpConnectionPtr& conn, const bitmq::basicCommonResponsePtr& message, muduo::Timestamp) {
        //1、找到信道
        Channel::ptr channel=_channel_manager->get(message->cid());
        if(channel.get()==nullptr)
        {
            DLOG("未找到相关信道!");
            return;
        }
        //2、将响应加入到 hash_map中
        channel->putbasicCommonResponse(message);
    }
    void consumerResponse(const muduo::net::TcpConnectionPtr& conn, const bitmq::basicConsumerResposePtr& message, muduo::Timestamp) {
       //1、找到信道
       Channel::ptr channel=_channel_manager->get(message->cid());
        if(channel.get()==nullptr)
        {
            DLOG("未找到相关信道!");
            return;
        }
       //2、封装异步任务，抛入到线程池
       _work->_pool.push([channel,message](){
            channel->consume(message);
       });
    }
    void onConnection(const muduo::net::TcpConnectionPtr&conn){
        if (conn->connected()) {
            _latch.countDown();//唤醒主线程中的阻塞
            _conn = conn;
        }else {
            //连接关闭时的操作
            _conn.reset();
        }
    }
    private:
       muduo::CountDownLatch _latch;//实现同步的
        muduo::net::EventLoopThread _loopthread;//异步循环处理线程,事件监控
        muduo::net::TcpConnectionPtr _conn;//客户端对应的连接
        muduo::net::TcpClient _connection;//客户端
        ProtobufDispatcher _dispatcher;//请求分发器
        ProtobufCodecPtr _codec;//协议处理器

        ChannelManager::ptr _channel_manager;
        AsyncWord::ptr _work;
};
}
#endif